package com.jielihaofeng.springboot.kafka.test;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author Johnnie Wind
 * @description 简单消费
 * @date 2021/3/3 14:26
 */
@Component
public class KafkaConsumer {

//    // 消费监听
//    @KafkaListener(topics = {"topic_wzy"})
//    public void onMessage1(ConsumerRecord<?, ?> record) {
//        System.out.println("简单消费:" + record.topic() + "-" + record.partition() + "-" + record.key() + ":" + record.value());
//        // 模拟dbus查询kafka日志
//        System.out.println("key:" + record.key());
//        System.out.println("offset:" + record.offset());
//        System.out.println(record.key() + "@@@" + record.value());
//    }
//
//    // 指定topic、partition、offset消费
//    // ①id：消费者ID；
//    // ② groupId：消费组ID；
//    // ③ topics：监听的topic，可监听多个；
//    // ④ topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
//    // onMessage2监听的含义：监听topic1的0号分区，同时监听topic2的0号分区和topic2的1号分区里面offset从8开始的消息。
//    @KafkaListener(
//            id = "consumer1",
//            groupId = "felix-group",
//            topicPartitions = {
//                    @TopicPartition(topic = "topic1", partitions = {"0"}),
//                    @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
//            })
//    public void onMessage2(ConsumerRecord<?, ?> record) {
//        System.out.println("简单消费:" + record.topic() + "-" + record.partition() + "-" + record.key() + ":" + record.value());
//    }
//
//    // 设置application.prpertise开启批量消费
//    @KafkaListener(id = "consumer2", groupId = "felix-group", topics = "topic1")
//    public void onMessage3(List<ConsumerRecord<?, ?>> records) {
//        System.out.println(">>>批量消费一次，records.size()=" + records.size());
//        for (ConsumerRecord<?, ?> record : records) {
//            System.out.println(record.value());
//        }
//    }
//
//    // 异常处理器 ConsumerAwareListenerErrorHandler
//    @Bean
//    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler(){
//        return (message,exception,consumer)->{
//            System.out.println("消费异常："+message.getPayload());
//            return null;
//        };
//    }
//
//    // 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
//    @KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
//    public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
//        throw new Exception("模拟异常");
//    }
//
//    // 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
//    @KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
//    public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
//        System.out.println("批量消费一次...");
//        throw new Exception("模拟异常");
//    }

}
